package com.yh.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName NIOServer
 * @Description TODO
 * @Author 0715-YuHao
 * @Date 2020/12/17 16:06
 */
public class NIOServer {

    private static Selector selector;
    private static LinkedBlockingQueue<SelectionKey> requestQueue;
    private static ThreadPoolExecutor threadPool;

    public static void main(String[] args) {
        init();
        listen();
    }

    private static void init() {
        ServerSocketChannel serverSocketChannel = null;
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress("localhost", 9000), 100);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException exception) {
            exception.printStackTrace();
        }
        requestQueue = new LinkedBlockingQueue<>(500);
        int processors = Runtime.getRuntime().availableProcessors();
        threadPool = new ThreadPoolExecutor(processors, processors * 2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(5));
        for (int i = 0; i < 10; i++) {
            threadPool.submit(new Worker());
        }
    }


    private static void listen() {
        while (true) {
            try {
                selector.select();
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();
                    handleRequest(key);
                }
            } catch (IOException exception) {
                exception.printStackTrace();
            }
        }
    }

    private static void handleRequest(SelectionKey key) {
        SocketChannel channel = null;
        try {
            if (key.isAcceptable()) {
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                channel = serverSocketChannel.accept();
                channel.configureBlocking(false);
                channel.register(selector, SelectionKey.OP_READ);
            }else if (key.isReadable()) {
                channel = (SocketChannel) key.channel();
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                int len = channel.read(buffer);
                if (len > 0) {
                    buffer.flip();
                    System.out.println("服务端收到请求：" + new String(buffer.array(), 0 ,len));
                    channel.register(selector, SelectionKey.OP_WRITE);
                }
            }else if (key.isWritable()) {
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                buffer.put("收到".getBytes());
                buffer.flip();
                channel = (SocketChannel) key.channel();
                channel.write(buffer);
                channel.register(selector, SelectionKey.OP_READ);
            }
        }catch (Exception e) {
            e.printStackTrace();
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException exception) {
                    exception.printStackTrace();
                }
            }
        }

    }

    private static class Worker implements Runnable{
        @Override
        public void run() {
            while (true) {
                try {
                    SelectionKey key = requestQueue.take();
                    handleRequest(key);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }

        private void handleRequest(SelectionKey key) {
            SocketChannel channel = null;
            try {
                if (key.isAcceptable()) {
                    System.out.println("[" + Thread.currentThread().getName() + "]收到请求");
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    channel = serverSocketChannel.accept();
                    channel.configureBlocking(false);
                    channel.register(selector, SelectionKey.OP_READ);
                }else if (key.isReadable()) {
                    channel = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int len = channel.read(buffer);
                    System.out.println("[" + Thread.currentThread().getName() + "]收到请求");
                    if (len > 0) {
                        buffer.flip();
                        System.out.println("服务端收到请求：" + new String(buffer.array(), 0 ,len));
                        channel.register(selector, SelectionKey.OP_WRITE);
                    }
                }else if (key.isWritable()) {
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    buffer.put("收到".getBytes());
                    buffer.flip();
                    channel = (SocketChannel) key.channel();
                    channel.write(buffer);
                    channel.register(selector, SelectionKey.OP_READ);
                }
            }catch (Exception e) {
                e.printStackTrace();
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (IOException exception) {
                        exception.printStackTrace();
                    }
                }
            }
        }
    }
}
